package chapter5.eg5;

import chapter5.AbstractConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.junit.Test;

public class StandaloneTest extends AbstractConsumer {

  @Test
  public void testConsume() {
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    List<TopicPartition> partitions = new ArrayList<>();
    List<PartitionInfo> allPartitions = consumer.partitionsFor("test-topic");
    if (null != allPartitions && !allPartitions.isEmpty()) {
      for (PartitionInfo partition : allPartitions) {
        partitions.add(new TopicPartition(partition.topic(), partition.partition()));
      }

      consumer.assign(partitions);
    }

    try {
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
        for (ConsumerRecord<String, String> record : records) {
          printRecord(record);
        }

        consumer.commitSync();
      }
    } catch (WakeupException e) {
      // ignore
    } finally {
      consumer.commitSync();
      consumer.close();
    }
  }

}
